package cn.itcast.nems.manager.rocketmq;

import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;

/**
 * <p>
 * 投递各种类型 mq消息
 * </p>
 *
 * @author zhouwei
 * @since 2023/10/26 2:05 PM
 **/
public class MessageManager {
    private static final Logger log = LoggerFactory.getLogger(MessageManager.class) ;

    /**
     * 异步投递顺序消息
     * @param  rocketMQTemplate 消息模版
     * @param destination 目标 tag
     * @param payload 消息体
     * @param hashKey 排序参考值
     **/
    public static void asyncSendOrderly(RocketMQTemplate rocketMQTemplate ,
                                        String destination, Object payload, String hashKey) {
        Message<Object> message = MessageBuilder
                .withPayload(payload)
                .setHeader(RocketMQHeaders.KEYS, hashKey)
                .build();
        rocketMQTemplate.asyncSendOrderly(destination, message, hashKey, new DefaultSendCallback(destination, payload));
    }
}

class DefaultSendCallback implements SendCallback {
    private static final Logger log = LoggerFactory.getLogger(MessageManager.class);
    private final String destination;
    private final Object payload;

    public DefaultSendCallback(String destination, Object payload) {
        this.destination = destination;
        this.payload = payload;
    }

    @Override
    public void onSuccess(SendResult sendResult) {
        log.info("【投递消息】投递成功，tag: {}，msgId：{}， 详情：{}", destination, sendResult.getMsgId(), payload);
    }

    @Override
    public void onException(Throwable throwable) {
        log.info("【投递消息】投递失败，tag: {}， 异常：{}， 异常详情：{}", destination, throwable.getMessage(), payload);
    }
}